package com.code.zg.rxapplication.ui;

import android.content.Intent;
import android.os.Bundle;
import android.support.v7.app.AppCompatActivity;
import android.util.Log;
import android.widget.Toast;

import com.code.zg.rxapplication.R;

import java.util.ArrayList;
import java.util.List;
import java.util.MissingFormatArgumentException;
import java.util.concurrent.TimeUnit;

import butterknife.ButterKnife;
import butterknife.OnClick;
import rx.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.functions.Action0;
import rx.functions.Func1;
import rx.functions.Func2;
import rx.subjects.AsyncSubject;
import rx.subjects.BehaviorSubject;
import rx.subjects.PublishSubject;
import rx.subjects.ReplaySubject;

/**
 * 1 创建 create from（迭代） just(整串) empty never error publishSubject BehaviorSubject ReplaySubject AsyncSubject
 * 2 过滤 就像一个纱窗 过滤掉不要的 留下想要的 filter（from 组合 过滤空数据）range repeat interval timer take skip frist last elementAt distinct distinctUtilChanged throttleFirst(指定时间的第一个) Timeout
 * 3 变换 就像一个加工厂 对原料进行加工 生成新的 map flamap (合并 允许交叉，重要是记住每一个发生错误的情况，flatmap会触发自己的onerror)
 *      concatMap (不合并 ，连续在一起 不会交叉) flatmapIterable 两两结对 switchMap停止前边的 scan将返回结果带回继续发射 groupBy cast buffer  window
 * 4 组合 merge (多输入 单输出 如果是同步的不会交叉 异步会交叉 mergeDelayError 发生错误所有完成后发送onerror)
 *      zip 多个输入进行函数处理后输出 有点像map 每个发射完才进行操作
 *      combineLastest 操作最近的两个 与zip不同在于任何一个发射了数据都可以工作
 *      and then when (rxjava-joins JoinObservable  类似zip)
 *      switch 取消订阅之前的 订阅新的
 *      startWith 和concat类似 只不过放在前边
 *
 *
 */
public class MainActivity extends AppCompatActivity {

    private static final String TAG = "rx";
    List<Integer> list=new ArrayList<>();
    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        ButterKnife.bind(this);
        list.add(1);
        list.add(2);
        list.add(3);
        list.add(3);
        list.add(4);
        list.add(5);
    }
    //eg 1 .simple emit
    @OnClick(R.id.btn1)
    void onBtn1Click() {
        Toast.makeText(this,"111",Toast.LENGTH_LONG).show();
        Observable<Integer> integerObservable = Observable.unsafeCreate(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i <5 ; i++) {
                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    subscriber.onNext(i);
                }
                subscriber.onCompleted();

            }
        });
        Subscription subscribe = integerObservable.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: "+e.getMessage());
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });

    }
    //eg2 list  array  迭代发送
    @OnClick(R.id.btn2) void onBtn2Click() {
//        List<Integer>list=new ArrayList<>();
//        list.add(1);
//        list.add(2);
//        list.add(3);
//        list.add(4);
//        Observable.from(list)
//                .subscribe(new Subscriber<Integer>() {
//                    @Override
//                    public void onCompleted() {
//                        Log.i(TAG, "onCompleted: ");
//                    }
//
//                    @Override
//                    public void onError(Throwable e) {
//                        Log.i(TAG, "onError: ");
//                    }
//
//                    @Override
//                    public void onNext(Integer integer) {
//                        Log.i(TAG, "onNext: "+integer);
//                    }
//                });
        Integer []s={1,2,3,4};
        Observable.from(s).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
    }

    private String hello(){
        return "hello";
    }
    private int number(){
        return 1000;
    }
    //function  在一个新的响应式框架的基础上迁移已经存在的代码
    @OnClick(R.id.btn3) void onBtn3Click() {
        //1
//        Observable.unsafeCreate(new Observable.OnSubscribe<String>() {
//            @Override
//            public void call(Subscriber<? super String> subscriber) {
//                subscriber.onNext(hello());
//                subscriber.onCompleted();
//            }
//        }).subscribe(new Subscriber<String>() {
//            @Override
//            public void onCompleted() {
//                Log.i(TAG, "onCompleted: ");
//            }
//
//            @Override
//            public void onError(Throwable e) {
//                Log.i(TAG, "onError: ");
//            }
//
//            @Override
//            public void onNext(String s) {
//                Log.i(TAG, "onNext: "+s);
//            }
//        });
        //2
//        Observable.just(hello()).subscribe(new Subscriber<String>() {
//            @Override
//            public void onCompleted() {
//                Log.i(TAG, "onCompleted: ");
//            }
//
//            @Override
//            public void onError(Throwable e) {
//                Log.i(TAG, "onError: ");
//            }
//
//            @Override
//            public void onNext(String s) {
//                Log.i(TAG, "onNext: "+s);
//            }
//        });
        //3 muti functions 依次执行
//        Observable.just(hello(),number()).subscribe(new Subscriber<Serializable>() {
//            @Override
//            public void onCompleted() {
//
//            }
//
//            @Override
//            public void onError(Throwable e) {
//
//            }
//
//            @Override
//            public void onNext(Serializable serializable) {
//                Log.i(TAG, "onNext: "+serializable.toString());
//            }
//        });
        //4 发射一整串
        Observable.just(list).subscribe(new Subscriber<List<Integer>>() {
            @Override
            public void onCompleted() {

            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(List<Integer> integers) {
                Log.i(TAG, "onNext: "+integers.toString());
            }
        });

    }


    @OnClick(R.id.btn4) void onBtn4Click() {
        Observable.empty().subscribe(new Subscriber<Object>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Object o) {
                Log.i(TAG, "onNext: "+o.toString());
            }
        });
        //啥也不干
        Observable.never().subscribe(new Subscriber<Object>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Object o) {
                Log.i(TAG, "onNext: "+o.toString());
            }
        });
        Observable.error(new MissingFormatArgumentException("aaa")).subscribe(new Subscriber<Object>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Object o) {
                Log.i(TAG, "onNext: "+o.toString());
            }
        });
    }

        //subject = observabe+observer   rxbus
    @OnClick(R.id.btn5) void onBtn5Click() {
        
        //当想要为公共资源创建独立抽象或更易观测的点时，这个是及其有用的  (同时可以被观测并且可以连接observables)
//        PublishSubject<String> publishSubject = PublishSubject.create();
//        publishSubject.subscribe(new Subscriber<String>() {
//            @Override
//            public void onCompleted() {
//                Log.i(TAG, "onCompleted: ");
//            }
//
//            @Override
//            public void onError(Throwable e) {
//                Log.i(TAG, "onError: ");
//            }
//
//            @Override
//            public void onNext(String s) {
//                Log.i(TAG, "onNext: "+s);
//            }
//        });
//        publishSubject.onNext("ssss");
        
//        public 的publishSubject 
        final PublishSubject<Boolean>publishSubject=PublishSubject.create();
        publishSubject.subscribe(new Subscriber<Boolean>() {
            @Override
            public void onCompleted() {
                
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Boolean aBoolean) {
                Log.i(TAG, "onNext:  dosomething "+aBoolean);
            }
        });
        //private observable
        Subscription subscribe = Observable.unsafeCreate(new Observable.OnSubscribe<Integer>() {
            @Override
            public void call(Subscriber<? super Integer> subscriber) {
                for (int i = 0; i < 5; i++) {
                    subscriber.onNext(i);
                    
                }
                subscriber.onCompleted();
            }
        }).doOnCompleted(new Action0() {
            @Override
            public void call() {
                Log.i(TAG, "call: ");
                publishSubject.onNext(true);
            }
        }).subscribe();
    }

    //先发送一个默认值或者订阅前最新的一个数据
    @OnClick(R.id.btn6) void onBtn6Click() {
        BehaviorSubject<Integer> integerBehaviorSubject = BehaviorSubject.create(1);
        integerBehaviorSubject.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "integerBehaviorSubject onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "integerBehaviorSubject onError: ");
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "integerBehaviorSubject onNext: "+integer);
            }
        });

        try {
            Thread.sleep(3000);
            integerBehaviorSubject.onNext(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

        //cache msg
    @OnClick(R.id.btn7) void onBtn7Click() {
        ReplaySubject<Integer> replaySubject = ReplaySubject.create();
        replaySubject.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
        replaySubject.onNext(1);
        replaySubject.onNext(3);
        replaySubject.onNext(4);
        replaySubject.onNext(5);
        replaySubject.onNext(6);

        replaySubject.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "replaySubject onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "replaySubject onError: ");
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "replaySubject onNext: "+integer);
            }
        });
    }

        //没反应。。。。
    @OnClick(R.id.btn8) void onBtn8Click() {
        AsyncSubject<Integer> asyncSubject = AsyncSubject.create();
        asyncSubject.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
        asyncSubject.subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted2: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext2: "+integer);
            }
        });
        asyncSubject.onNext(1);
        asyncSubject.onNext(2);
        asyncSubject.onNext(3);
    }


    @OnClick(R.id.btn9) void onBtn9Click() {
        Intent intent=new Intent(this,APPActivity.class);
        startActivity(intent);
    }
    @OnClick(R.id.btn11) void onBtn11Click() {
        Observable.just(hello()).repeat(3).subscribe(new Subscriber<String>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(String s) {
                Log.i(TAG, "onNext: "+s);
            }
        });
    }


    @OnClick(R.id.btn12) void onBtn12Click() {
//        defer 失败
    }


    @OnClick(R.id.btn13) void onBtn13Click() {
        Observable.range(10,3).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");

            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
    }


    @OnClick(R.id.btn14) void onBtn14Click() {
        Subscription subscribe = Observable.timer(3, TimeUnit.SECONDS).subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Long aLong) {
                Log.i(TAG, "onNext: " + aLong);
            }
        });
    Observable.interval(2,TimeUnit.SECONDS).subscribe(new Subscriber<Long>() {
        @Override
        public void onCompleted() {
            Log.i(TAG, "onCompleted: ");
        }

        @Override
        public void onError(Throwable e) {
            Log.i(TAG, "onError: ");
        }

        @Override
        public void onNext(Long aLong) {
            Log.i(TAG, "onNext: "+aLong);
        }
    });
    }


    @OnClick(R.id.btn15) void onBtn15Click() {
        //TODO implement
    }

        //filter------------------------------------------------------------------------
    @OnClick(R.id.btn16) void onBtn16Click() {
        //不行   应该发射单独的对象  针对每个对象进行判断
//        Observable.just(list).filter(new Func1<List<Integer>, Boolean>() {
//            @Override
//            public Boolean call(List<Integer> integers) {
//                List<Integer>data=new ArrayList<Integer>();
//                for (int i = 0; i < integers.size(); i++) {
//                    return integers.get(i)>3;
//                }
//                return true;
//            }
//        }).subscribe(new Subscriber<List<Integer>>() {
//            @Override
//            public void onCompleted() {
//                Log.i(TAG, "onCompleted: ");
//            }
//
//            @Override
//            public void onError(Throwable e) {
//                Log.i(TAG, "onError: ");
//            }
//
//            @Override
//            public void onNext(List<Integer>in) {
//                Log.i(TAG, "onNext: "+in);
//            }
//        });
        //最常用在于过滤空数据
        Observable.from(list).filter(new Func1<Integer, Boolean>() {
            @Override
            public Boolean call(Integer integer) {
                return integer>3;
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Integer aLong) {
                Log.i(TAG, "onNext: "+aLong);
            }
        });
    }

    //take takelast  skip skiplast first last firstordefault lastordefault elementAt elementAtordefault
    @OnClick(R.id.btn17) void onBtn17Click() {
        //TODO implement
        Observable.from(list).take(3).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
        Observable.from(list).takeLast(3).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
        Observable.from(list).elementAtOrDefault(5,100).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
    }

    //distinct distinctutilschanged
    @OnClick(R.id.btn18) void onBtn18Click() {
        Observable.from(list).distinct().subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
        Observable.from(list).distinctUntilChanged().subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
        Observable.from(list).lastOrDefault(0).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
    }

    //sample
    @OnClick(R.id.btn19) void onBtn19Click() {
        //最后一个
        Observable.from(list).sample(1,TimeUnit.MICROSECONDS).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
//只取10s钟的第一个  
        Observable.from(list).throttleFirst(10,TimeUnit.MILLISECONDS).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.i(TAG, "onNext: "+integer);
            }
        });
        Observable.never().timeout(2,TimeUnit.SECONDS).subscribe(new Subscriber<Object>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: "+e.getMessage());
            }

            @Override
            public void onNext(Object o) {
                Log.i(TAG, "onNext: "+o);
            }
        });
    }


    @OnClick(R.id.btn20) void onBtn20Click() {
        Observable.interval(2,TimeUnit.SECONDS).debounce(1,TimeUnit.SECONDS).subscribe(new Subscriber<Long>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Long aLong) {
                Log.i(TAG, "onNext: "+aLong);
            }
        });
    }
    //变换----------------------------------------------------------------------------------
    @OnClick(R.id.btn21)void onBtn21Click(){
        //对每一个值应用函数 并返回一下一个值
        Observable.just(1,3,2,5,4).scan(new Func2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) {
                if(integer>integer2){

                }
                return Math.max(integer,integer2);
            }
        }).subscribe(new Subscriber<Integer>() {
            @Override
            public void onCompleted() {
                Log.i(TAG, "onCompleted: ");
            }

            @Override
            public void onError(Throwable e) {
                Log.i(TAG, "onError: ");
            }

            @Override
            public void onNext(Integer aLong) {
                Log.i(TAG, "onNext: "+aLong);
            }
        });
    }
    //组合---------------------------------------------------------
    @OnClick(R.id.btn22)void onBtn22Click(){

    }
    @OnClick(R.id.btn23)void onBtn23Click(){

    }
    @OnClick(R.id.btn24)void onBtn24Click(){

    }
    @OnClick(R.id.btn25)void onBtn25Click(){

    }
    @OnClick(R.id.btn26)void onBtn26Click(){

    }
    @OnClick(R.id.btn27)void onBtn27Click(){

    }
    @OnClick(R.id.btn28)void onBtn28Click(){

    }
    @OnClick(R.id.btn29)void onBtn29Click(){

    }
    @OnClick(R.id.btn30)void onBtn30Click(){
        startActivity(new Intent(this,APP2Activity.class));
    }

    @Override
    protected void onDestroy() {
        super.onDestroy();
    }
}
